#include "StdAfx.h"
#include "MyDataReceiver.h"
#include "DataSClient.h"
#include "DataSClientDlg.h"

CCriticalSection CMyDataReceiver::m_sockCritSection;
bool CMyDataReceiver::m_connected		= false;
CString CMyDataReceiver::m_lastError	= "";
unsigned int CMyDataReceiver::m_socket	= -1;
CDataReceiver *CMyDataReceiver::pDataReceiver;
int CMyDataReceiver::m_tick_size;
int CMyDataReceiver::m_bidask_size;
int CMyDataReceiver::m_common_size;

CMyDataReceiver::CMyDataReceiver(void)
{
	m_port			= 0;
	m_connected		= false;
	m_socket		= -1;
	m_tick_size		= sizeof(STick);
	m_bidask_size	= 36;//sizeof(SBidAsk)-sizeof(double*)*2-sizeof(int*)*2;
	m_common_size	= sizeof(SCommon);
	pDataReceiver = new CDataReceiver;
}


CMyDataReceiver::~CMyDataReceiver(void)
{
	if(pDataReceiver->IsRunning())
		pDataReceiver->SuspendThread();
	delete pDataReceiver;
	Disconnect();
}

bool CMyDataReceiver::Connect(const char *ip, const unsigned int port)
{
	sockaddr_in dest_addr;
    HOSTENT *hst;
    char buff[1024];

	if(m_connected == true)
	{
		Disconnect();
	}

	if(pDataReceiver->IsRunning() == FALSE)
	{
		if(!pDataReceiver->InitMessageLoop())
		{
			m_lastError.Format("pDataReceiver->InitMessageLoop() error!");
			return false;
		}
	}

	m_port = port;
	m_ip = ip;

	if(m_ip.IsEmpty())
	{
		m_ip = "127.0.0.1";
	}

	if(WSAStartup(0x202, (WSADATA *)&buff[0]))
    {
		m_lastError.Format("WSAStart error %d\n", WSAGetLastError());
		return false;
    }

    m_socket = socket(AF_INET,SOCK_STREAM, 0);
    if(m_socket < 0)
    {
		m_lastError.Format("Socket() error %d\n", WSAGetLastError());
		return false;
    }

    dest_addr.sin_family = AF_INET;
    dest_addr.sin_port = htons(m_port);

    if(inet_addr(m_ip) != INADDR_NONE) 
	{
		dest_addr.sin_addr.s_addr = inet_addr(m_ip);
	}
    else
	{
		if(hst = gethostbyname(m_ip))
		{
			((unsigned long *)&dest_addr.sin_addr)[0] = ((unsigned long **)hst->h_addr_list)[0][0];
		}
		else 
		{
			m_lastError.Format("Invalid address %s\n", m_ip);
			closesocket(m_socket);
			WSACleanup();
			return false;
		}
	}

    if(connect(m_socket, (sockaddr *)&dest_addr, sizeof(dest_addr)))
    {
		switch(WSAGetLastError())
		{
		case WSAECONNREFUSED:
			m_lastError.Format("Connect error: refused by server (server offline) (%d)\n", WSAGetLastError());
			break;
		default:
			m_lastError.Format("Connect error %d\n", WSAGetLastError());
		}

		return false;
    }
	else
	{
		CString str;
		int size;
		char *data;
		char *ptr;
		int n;
		str.Format("%s;%d","ESZ2.CME;6EZ2.CME", 1);

		size=(int)str.GetLength()+NET_DATA_HEADER;
		data=new char[size];
		data[0]=NET_TYPE_LOGIN;
		data[1]=1;
		ptr=(char *)(&size);
		for(n=0;n<4;n++) data[n+2]=ptr[n];
		for(n=NET_DATA_HEADER;n<size;n++) data[n]=str[n-NET_DATA_HEADER];
		int nsize = send(m_socket, data, size, 0);
		m_connected = true;
		if(!pDataReceiver->IsRunning())
		{
			pDataReceiver->CreateThread();
		}
		Run();
		return true;
	}
}

void CMyDataReceiver::Run()
{
	_beginthread(RunThread, NULL, NULL);
}

void __cdecl CMyDataReceiver::RunThread(void *lpParameter)
{
	int bytes_recv, err;
	CString inText;
	timeval time_out;
	fd_set read_s;
	time_out.tv_sec = 0;
	time_out.tv_usec = 500000;
	char *ttext = NULL;
	size_t tlen = 0;
	char buff[MAX_NET_DATAPACKET];
	if(m_connected == false)
	{
		m_lastError = "CMyDataReceiver RunThread(): not connected!";
	}
	while(m_connected)
	{
		m_sockCritSection.Lock(); 
		FD_ZERO(&read_s);
		FD_SET(m_socket, &read_s);
		err = select(0, &read_s, NULL, NULL, &time_out);
		m_sockCritSection.Unlock();
		if(err == SOCKET_ERROR) 
		{
			m_connected = false;
			break;	
		}
		if((err != 0) && (FD_ISSET(m_socket, &read_s)))
		{
			m_sockCritSection.Lock();
			bytes_recv = recv(m_socket, &buff[0], sizeof(buff), 0);
			m_sockCritSection.Unlock();

			if(bytes_recv == SOCKET_ERROR)
			{
				m_connected = false;
				break;
			}
			if(bytes_recv > 0)
			{
				OnReceive(buff, bytes_recv);
				continue;
			}
		}
		Sleep(1000);
	}
}

void CMyDataReceiver::OnReceive(char *ptr, int nRead)
{
	static int cur_pos;
	static int data_size;
	static int type;
	static BOOL ret;
	static int i;
	static char *buff;

	ret = true;
	if(nRead>=MIN_NET_PACKETSIZE)
	{
		cur_pos=0;
		while(cur_pos<=nRead-MIN_NET_PACKETSIZE && ret)
		{
			data_size=*((int *)(ptr+2));
			type=ptr[0];
			
			if(cur_pos+data_size<=nRead && data_size>=MIN_NET_PACKETSIZE)
			{
				buff=new char[data_size];

				for(i=0;i<data_size;i++) buff[i]=*ptr++;

				// todo check if data size equal to expected
				if(type==NET_TYPE_TICK && data_size>m_tick_size+NET_DATA_HEADER) pDataReceiver->PostThreadMessage(WM_OnNewTickArrived,(WPARAM)buff,(LPARAM)data_size);
				else if(type==NET_TYPE_BIDASK && data_size>pDataReceiver->bidask_size+NET_DATA_HEADER) pDataReceiver->PostThreadMessage(WM_OnNewBidAskArrived,(WPARAM)buff,(LPARAM)data_size);
				else if(type==NET_TYPE_INTEREST && data_size>pDataReceiver->common_size+NET_DATA_HEADER) pDataReceiver->PostThreadMessage(WM_OnNewCommonArrived,(WPARAM)buff,(LPARAM)data_size);
				else if(type==NET_TYPE_INFO) pDataReceiver->PostThreadMessage(WM_OnNewMessageReceived,(WPARAM)buff,(LPARAM)data_size);
				else 
				{
					delete [] buff;
					pDataReceiver->Log("Invalid data received from socket.",IDM_ERROR,254);
					ret=FALSE;
				}

				cur_pos+=data_size;
			}
			else ret=FALSE;
		}
	}
}

void CMyDataReceiver::Disconnect()
{
	if(m_connected == true)
	{
		closesocket(m_socket);
		WSACleanup();
		m_connected = false;
	}
}

int StrLen(const char *Str)
{
	int MAX_STR_LEN = 2000;
	int i;
	int count = 0;
	if(Str == NULL)
	{
		return 0;
	}
	for(i=0; i<=MAX_STR_LEN; i++)
	{
		if(i == MAX_STR_LEN)
		{
			return 0;
		}
		if(Str[i] == '\0')
		{
			return count;
		}
		count++;
	}
	return 0;
}

unsigned int StringToUInt(const char *iStr)
{
	unsigned int m_delta_y;
	int i;
	unsigned int m_count = 1;
	unsigned int m_valI = 0;
	if(iStr == NULL)
	{
		return 0;
	}
	i = StrLen(iStr);
	if(i <= 0)
	{
		return m_valI;
	}
	for(i=i-1; i>=0; i--)
	{
		if(iStr[i] == '\0')
		{
			break;
		}
		m_delta_y = (unsigned int)(iStr[i]-'0');
		if(m_delta_y>9)
		{
			continue;
		}
		m_valI += m_count*m_delta_y;
		m_count*=10;
	}
	return m_valI;
}

const char *CMyDataReceiver::GetLastError()
{
	return m_lastError.GetString();
}

BOOL CMyDataReceiver::IsRunning()
{
	return m_connected;
}

/////////////////////////////////////////////////////////////////////////////
// CConnector

IMPLEMENT_DYNCREATE(CDataReceiver, CWinThread)

CDataReceiver::CDataReceiver()
{
	//Settings=theApp.SettingsDataReceiver;
	//Verbosity=theApp.Settings.verbosity;
	Status=TH_OFF;

	hTimer=NULL;
	hTimerQueue=NULL;
	Disconnecting=FALSE;
	Connecting=FALSE;
	ActiveCount=0;
	ModuleActionDone=NULL;
	ReceivedTicks=0;
	ReceivedBidAsks=0;
	ReceivedCommon=0;
	tick_size=sizeof(STick);
	bidask_size=36;//sizeof(SBidAsk)-sizeof(double*)*2-sizeof(int*)*2;
	common_size=sizeof(SCommon);

	for(int i=0;i<PL_COUNT;i++) data_pending[i]=FALSE;
}

CDataReceiver::~CDataReceiver()
{
	DeInitMessageLoop();
}

void CDataReceiver::Log(LPCTSTR message, BYTE type, int code)
{
	if(message==NULL || !(1<<(type-1) & Verbosity)) return;

	//SLogEvent* LogEvent=new SLogEvent;
	//LogEvent->code=code;
	//LogEvent->time=theApp.CurrentTime;
	//LogEvent->module=ID_DATARECEIVER;
	////LogEvent->module_id="";
	////LogEvent->ticker_name="";
	//LogEvent->type=type;
	//LogEvent->message=message;

	//if(theApp.ControlServer) theApp.ControlServer->PostThreadMessage(WM_OnLogMessage, (WPARAM)LogEvent, 0);
}

BOOL CDataReceiver::InitInstance()
{
    return TRUE;
}

BOOL CDataReceiver::InitMessageLoop()
{
	CString str;
	BOOL Failed=FALSE;

	NewMessage=NULL;
	LastMessagePtr=NULL;
	FirstMessagePtr=NULL;
	CurMessagePtr=NULL;
	MessageLoopSize=0;

	if(InitializeCriticalSectionAndSpinCount(&LoopLock,4000))
	{
		EnterCriticalSection(&LoopLock);

		MessageLoopSize=EVENT_LOOP_SIZE_HUGE;
		FirstMessagePtr=new SMessage[MessageLoopSize];

		if(FirstMessagePtr)
		{
			LastMessagePtr=FirstMessagePtr+MessageLoopSize-1;
			LastMessagePtr->Status=2;
			LastMessagePtr=FirstMessagePtr;
			CurMessagePtr=FirstMessagePtr;
		}
		else Failed=TRUE;

		LeaveCriticalSection(&LoopLock);
	}
	else Failed=TRUE;

	if(Failed) 
	{
		DeInitMessageLoop();
		Log("Failed to initialize message loop.",IDM_ERROR,15);
	}

	return !Failed;
}

void CDataReceiver::DeInitMessageLoop()
{
	BOOL Initialized;

	if(MessageLoopSize) Initialized=TRUE;
	else Initialized=FALSE;

	if(Initialized) EnterCriticalSection(&LoopLock);

	if(NewMessage) CloseHandle(NewMessage);
	if(FirstMessagePtr) delete [] FirstMessagePtr;

	NewMessage=NULL;
	FirstMessagePtr=NULL;
	LastMessagePtr=NULL;
	CurMessagePtr=NULL;
	MessageLoopSize=0;

	if(Initialized) DeleteCriticalSection(&LoopLock);
}

BOOL CDataReceiver::PostThreadMessage(UINT message, WPARAM wParam, LPARAM lParam)
{
	if(MessageLoopSize) EnterCriticalSection(&LoopLock);
	if(!MessageLoopSize) return FALSE;

	if(CurMessagePtr->Status!=0) // waiting for empty row
	{
		Log("Message buffer overflow detected.",IDM_ALERT,2001);

		for(int i=0;i<50 && CurMessagePtr->Status!=0;i++) Sleep(OPER_SLEEP_MS);
	}

	if(CurMessagePtr->Status==0)
	{
		CurMessagePtr->Message=message;
		CurMessagePtr->wParam=wParam;
		CurMessagePtr->lParam=lParam;
		CurMessagePtr->Status=1;

		CurMessagePtr++;
		if(CurMessagePtr->Status==2) CurMessagePtr=FirstMessagePtr;

		SetEvent(NewMessage);
		LeaveCriticalSection(&LoopLock);

		return TRUE;
	}
	else 
	{
		LeaveCriticalSection(&LoopLock);
		return FALSE;
	}
}

int CDataReceiver::Run()
{
	BOOL NewMessageExist=FALSE;
	BOOL finalized=FALSE;
	UINT message;
	MSG m_msgCur;
	Status==TH_ON;
	// main processing cycle
	while(1)
	{		
		while(PeekMessage(&m_msgCur, (HWND)NULL, NULL, NULL, PM_REMOVE)) 
			DispatchMessage(&m_msgCur);

		if(NewMessageExist || WaitForSingleObject(NewMessage,0)==WAIT_OBJECT_0)
		{
			NewMessageExist=FALSE;

			while(LastMessagePtr->Status==1) 
			{
				message=LastMessagePtr->Message;

				if(message==WM_OnNewTickArrived) 
				{
					buff=(char *)LastMessagePtr->wParam;
					OnNewTickArrived(buff);
					delete [] buff;
				}
				else if(message==WM_OnNewBidAskArrived) 
				{
					buff=(char *)LastMessagePtr->wParam;
					OnNewBidAskArrived(buff);
					delete [] buff;
				}
				else if(message==WM_OnNewCommonArrived) 
				{
					buff=(char *)LastMessagePtr->wParam;
					OnNewCommonArrived(buff);
					delete [] buff;
				}
				else if(message==WM_OnNewMessageReceived) 
				{
					buff=(char *)LastMessagePtr->wParam;
					OnNewMessageReceived(buff,(int)LastMessagePtr->lParam);
					delete [] buff;
				}
				else if(message==WM_OnDisconnect) 
				{
					Connecting=FALSE;
					Disconnecting=TRUE;
				}
				else if(message==WM_QUIT) return ExitInstance();	

				LastMessagePtr->Message=0;
				LastMessagePtr->lParam=0;
				LastMessagePtr->wParam=0;
				LastMessagePtr->Status=0;
				LastMessagePtr++;
				if(LastMessagePtr->Status==2) LastMessagePtr=FirstMessagePtr;
			} 
		}

		for(n=0;n<PL_COUNT;n++)
		{
			//if(data_pending[n] && abs(theApp.CurrentTime-data_time[n])>=theApp.SettingsGateways[n].processing_idle)
			//{
			//	theApp.DataManager->PostThreadMessage(WM_OnDataFinalize,(WPARAM)n,0);
			//	data_pending[n]=FALSE;
			//	finalized=TRUE;
			//}
		}
		
		if(!finalized)
		{
			if(WaitForSingleObject(NewMessage,RUN_SLEEP_MS)==WAIT_OBJECT_0) NewMessageExist=TRUE;
		}
		else finalized=FALSE;
	}
}

void CDataReceiver::OnNewTickArrived(char *buffer)
{
	//if(!IsRunning() || !Settings.allow_ticks) return;

	//if(buffer[1]!=1) return;

	//if(buffer[NET_DATA_HEADER]==0 || !theApp.TickersCodeStr.Lookup(buffer+NET_DATA_HEADER,(void*&)ticker_info)) return;

	ReceivedTicks++;
	//cur=NET_DATA_HEADER+1;

	//while(buffer[cur++]!=0);
	//
	//tick=new STick;
	//target=(char *)tick;
	//for(n=0;n<tick_size;n++) target[n]=buffer[cur++];

	//platform=tick->platform;
	//delay=theApp.CurrentTime-tick->time;
	//tick->delay=delay;
	//tick->flags=tick->flags | IDD_REMOTE_DATA;

	//theApp.DataManager->PostThreadMessage(WM_OnNewTickArrived,(WPARAM)tick,(LPARAM)ticker_info);

	//if(!theApp.SettingsGateways[platform].allow_gateway) CalcDelays(&DataDelays[platform], delay, theApp.SettingsGateways[platform].calc_avg_period, 0);

	//if(!data_pending[platform])
	//{
	//	data_pending[platform]=TRUE;
	//	data_time[platform]=theApp.CurrentTime;
	//}
}

void CDataReceiver::OnNewBidAskArrived(char *buffer)
{
	//if(!IsRunning() || !Settings.allow_bidask) return;

	//if(buffer[1]!=1) return;

	//if(buffer[NET_DATA_HEADER]==0 || !theApp.TickersCodeStr.Lookup(buffer+NET_DATA_HEADER,(void*&)ticker_info)) return;

	//ReceivedBidAsks++;
	//cur=NET_DATA_HEADER+1;

	//while(buffer[cur++]!=0);

	//bidask=new SBidAsk;
	//target=(char *)bidask;
	//for(n=0;n<bidask_size;n++) target[n]=buffer[cur++];

	//if(bidask->askrows>0)
	//{
	//	bidask->ask=new double[bidask->askrows];
	//	bidask->asksize=new int[bidask->askrows];

	//	target=(char *)bidask->ask;
	//	len=bidask->askrows*sizeof(double);
	//	for(n=0;n<len;n++) target[n]=buffer[cur++];

	//	target=(char *)bidask->asksize;
	//	len=bidask->askrows*sizeof(int);
	//	for(n=0;n<len;n++) target[n]=buffer[cur++];
	//}

	//if(bidask->bidrows>0)
	//{
	//	bidask->bid=new double[bidask->bidrows];
	//	bidask->bidsize=new int[bidask->bidrows];

	//	target=(char *)bidask->bid;
	//	len=bidask->bidrows*sizeof(double);
	//	for(n=0;n<len;n++) target[n]=buffer[cur++];

	//	target=(char *)bidask->bidsize;
	//	len=bidask->bidrows*sizeof(int);
	//	for(n=0;n<len;n++) target[n]=buffer[cur++];
	//}

	//platform=bidask->platform;
	//bidask->flags=bidask->flags | IDD_REMOTE_DATA;

	//if(bidask->ticknum==0) 
	//{
	//	bidask->index=++ticker_info->last_bidask.index;
	//	bidask->ticknum=ticker_info->last_bidask.ticknum;
	//}

	//theApp.DataManager->PostThreadMessage(WM_OnNewBidAskArrived,(WPARAM)bidask,(LPARAM)ticker_info);

	//if(!data_pending[platform])
	//{
	//	data_pending[platform]=TRUE;
	//	data_time[platform]=theApp.CurrentTime;
	//}
}

void CDataReceiver::OnNewCommonArrived(char *buffer)
{
	//if(!IsRunning() || !Settings.allow_common) return;

	//if(buffer[1]!=1) return;

	//if(buffer[NET_DATA_HEADER]==0 || !theApp.TickersCodeStr.Lookup(buffer+NET_DATA_HEADER,(void*&)ticker_info)) return;

	//ReceivedCommon++;
	//cur=NET_DATA_HEADER+1;

	//while(buffer[cur++]!=0);

	//common=new SCommon;
	//target=(char *)common;
	//for(n=0;n<common_size;n++) target[n]=buffer[cur++];

	//platform=common->platform;
	//common->flags=common->flags | IDD_REMOTE_DATA;

	//theApp.DataManager->PostThreadMessage(WM_OnNewCommonArrived,(WPARAM)common,(LPARAM)ticker_info);

	//if(!data_pending[platform])
	//{
	//	data_pending[platform]=TRUE;
	//	data_time[platform]=theApp.CurrentTime;
	//}
}

void CDataReceiver::OnNewMessageReceived(char *message, int length)
{
	if(message) delete [] message;
}

BOOL CDataReceiver::IsRunning()
{
	if(Status==TH_ON) return TRUE;
	else return FALSE;
}
